package com.qf.qfdrivermqtt.config;

import com.alibaba.fastjson.JSON;
import com.qf.common.constant.KafkaConstant;
import com.qf.common.feigns.DeviceTopicApi;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.nio.channels.Channel;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;

@Configuration
@IntegrationComponentScan
public class MqttConfig implements InitializingBean {

    @Autowired
    private DeviceTopicApi deviceTopicApi;

    private String[] inTopics = null;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Override
    public void afterPropertiesSet() throws Exception {
        //查询所有设备主题
        List<String> topics = deviceTopicApi.findAllTopics();

        //从所有主题中获取出入站相关的主题,实现入站的监听
        inTopics = topics.stream().filter(topic->{
            return topic.startsWith("/qfjava/device/data");
        }).collect(Collectors.toList()).toArray(new String[0]);

    }

    @Bean
    public MessageChannel inChannel(){
        return new DirectChannel();
    }

    @Bean
    public MessageProducer messageProducer(){
        String clientId = "qf-driver-mqtt_in";
        MqttPahoClientFactory connFactory = clientFactory();
        String[] topics = inTopics;
        MqttPahoMessageDrivenChannelAdapter mp = new MqttPahoMessageDrivenChannelAdapter(clientId,connFactory,topics);

        mp.setConverter(new DefaultPahoMessageConverter());
        mp.setQos(1);
        //将入站的消息转到通道中
        mp.setOutputChannel(inChannel());

        return mp;
    }

    //消费入站消息
    @Bean
    @ServiceActivator(inputChannel = "inChannel")
    public MessageHandler inMessageHandler(){
        return msg->{
            System.out.println("--------------------------" + msg);
            //将收集到的入站消息通过消息中间件进行异步提速+削峰填谷（kafka）
            //发送消息到kafka消息中间件

            String errorJson = JSON.toJSONString(msg.getPayload());
            System.out.println("-------------------------- " + errorJson);

            String json = msg.getPayload().toString();
            System.out.println("====" + json);


            ProducerRecord record = new ProducerRecord(KafkaConstant.DEVICE_DATA_TOPIC,json);
            try {
                kafkaTemplate.send(record);
            }catch (Exception e){
                e.printStackTrace();
                //重试
            }
        };
    }

    @Bean
    public MessageChannel outChannel(){
        return new DirectChannel();
    }

    //出站消息处理器
    @Bean
    @ServiceActivator(inputChannel = "outChannel")
    public MessageHandler outMessageHandler(){
        String clientId = "qf-driver-mqtt_out";
        MqttPahoClientFactory connFactory = clientFactory();
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId,connFactory);

        //设置默认的服务质量等级
        messageHandler.setDefaultQos(1);
        //设置默认主题
        messageHandler.setDefaultTopic("qf-driver-mqtt/default");
        //支持异步发送（效率更高）
        messageHandler.setAsync(true);

        return messageHandler;
    }

    @Bean
    public MqttPahoClientFactory clientFactory(){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();

        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{"tcp://localhost:1883"});
        options.setUserName("qf");
        options.setPassword("qf".toCharArray());

        options.setCleanSession(false);

        factory.setConnectionOptions(options);
        return factory;
    }

}
